package cn.itcast.flink.join;

import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.windows.GlobalWindow;
import org.apache.flink.util.Collector;

import java.math.BigDecimal;
import java.util.Arrays;

/**
 * Author itcast
 * Date 2022/1/14 11:13
 * Desc 统计五位同学的平均成绩，增量还是全量的？ 全量，对窗口内的数据进行一次计算。
 * 分析：
 * 1.需要使用计数窗口 countWindow = 5
 * 2.每五位同学的总成绩和个数 总成绩/个数
 * 3.打印输出
 */
public class CountWindowProcessDemo {
    public static void main(String[] args) throws Exception {
        //获取流执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //设置并行度
        env.setParallelism(1);
        //获取数据源
        DataStreamSource<Tuple3<String,String,Long>> source = env.fromCollection(Arrays.asList(ENGLISH));

        //不分组，开窗每5个分一个组，process全量计算平均值
        /*source.countWindowAll(3)
                .process(new ProcessAllWindowFunction<Tuple3<String, String, Long>, Double, GlobalWindow>() {
                    @Override
                    public void process(Context context, Iterable<Tuple3<String, String, Long>> elements, Collector<Double> out) throws Exception {
                        //统计所有的同学每个的平均值
                        Long sumScore = 0L;
                        //Iterable 迭代器中保存的就是窗口中的这3个值
                        for (Tuple3<String, String, Long> element: elements){
                            sumScore += element.f2;
                        }
                        Double avg = (sumScore / 3.0D);
                        out.collect(avg);
                    }
                }).print();*/
        //分组，开窗每5个分一个组，process全量计算
        source.keyBy(t->t.f0)
                .countWindow(3)
                //全量计算 process apply
                //@param <IN> The type of the input value.
                //@param <OUT> The type of the output value.
                //@param <KEY> The type of the key.
                //@param <W> The type of {@code Window} that this window function can be applied on.
                .process(new ProcessWindowFunction<Tuple3<String, String, Long>, Double, String, GlobalWindow>() {
                    // process
                    @Override
                    public void process(String clazz, Context context, Iterable<Tuple3<String, String, Long>> elements, Collector<Double> out) throws Exception {
                        //统计所有的同学每个的平均值
                        Long sumScore = 0L;
                        //Iterable 迭代器中保存的就是窗口中的这3个值
                        for (Tuple3<String, String, Long> element: elements){
                            sumScore += element.f2;
                        }
                        BigDecimal bigDecimal = new BigDecimal(sumScore / 3.0D).setScale(2,BigDecimal.ROUND_HALF_UP);
                        out.collect(bigDecimal.doubleValue());
                    }
                }).print();
        //打印输出结果
        //执行流环境
        env.execute();
    }

    public static final Tuple3[] ENGLISH = new Tuple3[]{
            Tuple3.of("class1", "张三", 100L),
            Tuple3.of("class1", "李四", 78L),
            Tuple3.of("class1", "王五", 99L),
            Tuple3.of("class2", "赵六", 81L),
            Tuple3.of("class2", "小七", 59L),
            Tuple3.of("class2", "小八", 97L),
            Tuple3.of("class2", "小九", 98L),
            Tuple3.of("class2", "小时", 98L)
    };
}
